#include "EventLoop.h"
#include "Channel.h"
#include "Connection.h" // for delete connection
#include "Socket.h"
#include "BaseThread.h" // for GetCurrentThreadId
#include "IMultiplex.h"
#include "Wakeup.h"
#include "Options.h"
#include "TimerQueue.h"

#ifdef _MSC_VER
#pragma warning(disable:4355) // for 'this' pointer used in base member initializer list
#endif
namespace dsth {

EventLoop::EventLoop() 
    : looping_(false),
    stop_(true),
    polling_(false),
    thread_id_(0),
    multiplexor_(CreateMultiplexor()),
    mutex_(),
    wakeup_(new Wakeup(this)),
    timer_queuer_(new TimerQueue(this)) {
}

EventLoop::~EventLoop() {
}

void EventLoop::Loop() {
    assert(!looping_);
    if (looping_) {
        return;
    }
    wakeup_->AttachToLoop();
    thread_id_ = BaseThread::GetCurrentThreadId();
    stop_ = false;
    looping_ = true;
    int timeout = GetOption("ReactorTimeoutUsec").intVal;
    while (!stop_) {
        channel_list_.clear();
        polling_ = true;
        Timestamp recv_time = multiplexor_->Run(timeout, &channel_list_);
        polling_ = false;
        for (IMultiplex::ChannelList::iterator it = channel_list_.begin();
            it != channel_list_.end(); ++it) {
            (*it)->HandleEvent(recv_time);
        }
        RunAllTasks();
    }
}

void EventLoop::Quit() {
    stop_ = true;
    SelfWaken();
}

void EventLoop::SelfWaken() {
    if (polling_) {
        wakeup_->Waken();
    }
}

void EventLoop::RunInLoop(const EventTask& task) {
    if (IsInLoopThread() || !looping_) {
        task();
    }
    else {
        MutexLock lock(&mutex_);
        tasks_ += task;
        SelfWaken();
    }
}

bool EventLoop::IsInLoopThread() {
    return thread_id_ == BaseThread::GetCurrentThreadId();
}

void EventLoop::RunAllTasks() {
    EventTask calling_task;
    {
        MutexLock lock(&mutex_);
        calling_task = tasks_;
        tasks_.Clear();
    }
    calling_task();
    // delete erased connections.
    // Actually, this's a temporary solution to destroy erased connections.
    // Need be fixed using scope point or something more graceful.
    while (!conn_list_.empty()) {
        Connection* conn = conn_list_.back();
        conn_list_.pop_back();
        delete conn;
    }
    assert(conn_list_.empty());
}

void EventLoop::Update(Channel* channel) {
    assert(channel->GetOwnerLoop() == this);
    // Loop may not looping when update channel.
    assert(!looping_ || IsInLoopThread());
    multiplexor_->UpdateChannel(channel);
}

void EventLoop::Remove(Channel* channel) {
    assert(channel->GetOwnerLoop() == this);
    assert(IsInLoopThread());
    multiplexor_->RemoveChannel(channel);
}

void EventLoop::AddErasedConnection(Connection* conn) {
    assert(IsInLoopThread());
    assert(conn);
    conn_list_.push_back(conn);
}

#ifdef IS_WINDOWS
TimerId EventLoop::RunAt(const Timestamp& when, const TimerCallBack& cb) {
    int64_t time_diff = TimeDifference(when, Timestamp::GetNowTime());
    if (time_diff > 0) {
        // is milli seconds
        return timer_queuer_->AddTimer(cb, static_cast<uint64_t>(time_diff / 1000), 0);
    }
    return 0;
}

TimerId EventLoop::RunAfterMilliSeconds(uint64_t duetime, const TimerCallBack& cb) {
    return timer_queuer_->AddTimer(cb, duetime, 0);
}

TimerId EventLoop::RunPeriodMilliSeconds(uint64_t period, const TimerCallBack& cb) {
    return timer_queuer_->AddTimer(cb, period, period);
}
#else // POSIX
// TODO: for POSIX
#endif

} // namespace
